package com.sinopec.pi.sparkguide

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 查看 RDD Cache 运行时的世系
 */
object CachePersist {
    def main(args: Array[String]): Unit = {
        // 指定 Master 为 本地两核，支持多种部署运行模式
        val conf = new SparkConf().setAppName("Spark Pi").setMaster("local[2]")
        val sc = new SparkContext(conf)

        // 英文文本，按空格分割，生成单词对，统计单词出现次数
        // 获取本地文件的路径
        val filePath = WordCount.getClass.getClassLoader.getResource("nsdi_spark.txt").toURI.getPath
        // 支持 hdfs:// 和 file://
        val fileRDD: RDD[String] = sc.textFile(filePath)

        val wcRdd: RDD[(String, Int)] = fileRDD.flatMap(line => line.split(" "))
            .map(word => (word, 1))
            // Scala 类的隐式转换，调用的不是 RDD 的方法，而是 PairRDDFunction 的方法, 见 object RDD 提供的隐式转换
            .reduceByKey((a, b) => a + b)

        wcRdd.persist(StorageLevel.MEMORY_ONLY)
        wcRdd.count()

        // 当再次执行时，查看 Stage，会从内存读取
        wcRdd.count()

        // wait to see the web ui
        Thread.sleep(1000000)
    }
}
